package com.halflife.consumers;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;


/**
 * Kafka消费者和消费者组
 */
public class ConsumerInfo {

    public static void main(String[] args) {
        String topic = "mock1";
        String group = "group6767";
        Properties props = new Properties();
        props.put("bootstrap.servers", "master:9091,master:9092,master:9093");
        /*指定分组ID*/
        props.put("group.id", group);
        props.put(ConsumerConfig.CLIENT_ID_CONFIG, "client001");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);



        try {
            List<PartitionInfo> list =  consumer.partitionsFor(topic);
            List<TopicPartition> topicPartitions = new ArrayList<>();
            for (PartitionInfo partitionInfo : list) {
                TopicPartition topicPartition = new TopicPartition(partitionInfo.topic(), partitionInfo.partition());
                topicPartitions.add(topicPartition);
            }
            consumer.assign(topicPartitions);
            for (TopicPartition topicPartition : topicPartitions) {
                long offset = consumer.position(topicPartition);
                System.out.println("==============================================");
                System.out.println(topicPartition.topic()+":"+topicPartition.partition()+"---"+offset);
            }
//            System.out.println(list);
        } finally {
            consumer.close();
        }

    }
}
